package com.foo.DataSender;

import java.util.Properties;

import javax.ejb.Stateless;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;

/**
 * Session Bean implementation class DataSender
 */
@Stateless
public class DataSender implements DataSenderRemote, DataSenderLocal {

    /**
     * Default constructor. 
     */
    public DataSender() 
    {
    }

	Topic topic = null;
	TopicConnectionFactory factory = null;
	TopicConnection connection = null;
	MessageProducer producer = null;
	MessageProducer snapshotproducer = null;
	MessageProducer temporalproducer = null;
	MessageProducer analyzedproducer = null;
	MessageProducer realtimeproducer = null;
	MessageProducer historicproducer = null;
	TopicSession session = null;
	Topic snapshottopic =null; 
	Topic temporaltopic = null;
	Topic analyzedtopic = null;
	Topic realtimetopic =null;
	Topic historictopic =null;
	
	Queue queue = null;
	QueueConnectionFactory qfactory = null;
	QueueConnection qconnection = null;
	QueueSender qsender = null;
	QueueSession qsession = null;
	
	@Override
	public void sendMessage(String operation,String data) //TickerInfoData[] data
	{
		try 
		{
			if(operation.equalsIgnoreCase("LoadHistoricalData"))
			{
				//Load Historical Data
				QueueConnectionFactory qfactory = null;
				Properties props = new Properties();
				props.put(Context.PROVIDER_URL, "jnp://localhost:1099");
				props.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.interfaces.NamingContextFactory");
				props.put(Context.URL_PKG_PREFIXES,"org.jboss.naming:org.jnp.interfaces");
				Context ctx = new InitialContext(props);
				qfactory = (QueueConnectionFactory) ctx.lookup("XAConnectionFactory");
				qconnection = qfactory.createQueueConnection();
				qsession = qconnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
				queue = (Queue) ctx.lookup("queue/HistoricalDataStorage");
				qsender = qsession.createSender(queue);
				//create and set a message to send
				TextMessage message = qsession.createTextMessage(data);
				message.setJMSExpiration(5000);
				qsender.send(message);
				qsession.close ();
				qconnection.close();
			}
			else //Topics
			{
				Properties props = new Properties();
				props.put(Context.PROVIDER_URL, "jnp://localhost:1099");
				props.put(Context.INITIAL_CONTEXT_FACTORY,"org.jnp.interfaces.NamingContextFactory");
				props.put(Context.URL_PKG_PREFIXES,"org.jboss.naming:org.jnp.interfaces");
				Context ctx = new InitialContext(props);
				factory = (TopicConnectionFactory) ctx.lookup("XAConnectionFactory");
				connection = factory.createTopicConnection();
				
				session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
				
				/*qfactory = (QueueConnectionFactory) ctx.lookup("XAConnectionFactory");
				qconnection = qfactory.createQueueConnection();
				qsession = qconnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);*/
				
				if(operation.equals("Spatial"))
				{
					topic = (Topic) ctx.lookup("topic/MySpatialTopic");
					producer = session.createProducer(topic);
					TextMessage message = session.createTextMessage();
					message.setText(data);
					producer.send(message);
					System.out.println("Sending message");
					session.close();
					connection.close();
				}
				else if (operation.equals("Snapshot"))
				{
					snapshottopic = (Topic) ctx.lookup("topic/MySnapshotTopic");
					snapshotproducer = session.createProducer(snapshottopic);
					TextMessage message = session.createTextMessage();
					message.setText(data);
					//System.out.println("Created text message to send "+message);	
					snapshotproducer.send(message);	
					System.out.println("Sending message");
					session.close();
					connection.close();
				}
				else if(operation.equals("Temporal"))
				{
					temporaltopic = (Topic) ctx.lookup("topic/MyTemporalTopic");
					temporalproducer = session.createProducer(temporaltopic);
					TextMessage message = session.createTextMessage();
					message.setText(data);
					//System.out.println("Created text message to send "+message);	
					temporalproducer.send(message);	
					System.out.println("Sending message");
					session.close();
					connection.close();
				}
				else if(operation.equals("Analyzed"))
				{
					analyzedtopic = (Topic) ctx.lookup("topic/MyAnalyzedTopic");
					analyzedproducer = session.createProducer(analyzedtopic);
					TextMessage message = session.createTextMessage();
					message.setText(data);
					//System.out.println("Created text message to send "+message);	
					analyzedproducer.send(message);	
					System.out.println("Sending message");
					session.close();
					connection.close();
				}
				else if (operation.equals("RealTime"))
				{
					realtimetopic = (Topic) ctx.lookup("topic/MyRealTimeTopic");
					realtimeproducer = session.createProducer(realtimetopic);
					TextMessage message = session.createTextMessage();
					message.setText(data);
					//System.out.println("Created text message to send "+message);	
					realtimeproducer.send(message);	
					System.out.println("Sending message");
					session.close();
					connection.close();
					
				}
				else if (operation.equals("HistoricData"))
				{
					historictopic = (Topic) ctx.lookup("topic/MyHistoricData");
					historicproducer = session.createProducer(historictopic);
					TextMessage message = session.createTextMessage();
					message.setText(data);
					//System.out.println("Created text message to send "+message);	
					historicproducer.send(message);	
					System.out.println("Sending message");
					session.close();
					connection.close();						
				}
			}
		} 
		catch (Exception e) {
			e.printStackTrace();
		}
	}
	/*public static void main(String... args)
	{
		DataSender client = new DataSender();
	}*/
}


